package com.dapeng.reactor;

import com.dapeng.debug.Log;

import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.ServerSocketChannel;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;

public class SelectorGroup {

    private AtomicInteger idx = new AtomicInteger(0);

    private SelectorThread[] selectorThreads;

    private SelectorGroup workGroup;

    public SelectorGroup(int num, String threadName) {
        selectorThreads = new SelectorThread[num];
        for (int i = 0; i < num; i++) {
            selectorThreads[i] = new SelectorThread(this);
            new Thread(selectorThreads[i], threadName + i).start();
        }

        workGroup = this;
    }

    public void bind(int port) {
        try {
            ServerSocketChannel serverChannel = ServerSocketChannel.open();
            serverChannel.configureBlocking(false);
            serverChannel.bind(new InetSocketAddress(port));

            Log.info("服务端启动，地址：{}", serverChannel.getLocalAddress());

            registerSelect(serverChannel);
        } catch (Exception e) {
            Log.error(e);
        }
    }

    public void setWork(SelectorGroup workGroup) {
        this.workGroup = workGroup;
    }

    public void registerSelect(Channel channel) {
        SelectorThread selectorThread = getRegisterSelector(channel);
        selectorThread.addChannelQueue(channel);
        selectorThread.getSelector().wakeup();
    }

    private SelectorThread getRegisterSelector(Channel channel) {
        SelectorThread[] localSelectorThread;
        if (channel instanceof ServerSocketChannel) {
            localSelectorThread = this.selectorThreads;
            return localSelectorThread[idx.incrementAndGet() % localSelectorThread.length];
        }

        localSelectorThread = this.workGroup.selectorThreads;
        Arrays.sort(localSelectorThread);

        return localSelectorThread[0];
    }

}
